/*******************************************************************************
 * Copyright (c) 2012, 2017 IBM Corp.
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Eclipse Distribution License v1.0 which accompany this distribution. 
 *
 * The Eclipse Public License is available at 
 *   http://www.eclipse.org/legal/epl-v10.html
 * and the Eclipse Distribution License is available at 
 *   http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * Contributors:
 *    Ian Craggs - initial contribution
 *******************************************************************************/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTClient.h"
#include <pthread.h>
#include "cJSON.h"

 #include "msg_queue_peer.h"
 #include "shmem.h"
//#include "xulihua.h"
#define ADDRESS     "tcp://192.168.1.51:1883"
#define CLIENTID    "ExampleClientSub"
#define TOPIC       "down"
#define TOPIC1       "up"
#define PAYLOAD     "Hello World!"
#define QOS         1
#define TIMEOUT     10000L
int stm=0;
//共享内存
static struct shm_param para;

union val_t
{ 
    int b_val;  //bool类型存储空间
    int i_val;   //整形值存储空间
    float f_val;  //浮点值存储空间
};

struct student
{
	char str[64];  // 类似放设备名字的或地址
    int key;  //唯一键值
	int type;  //数据点类型
    int dev_type;  //数据点属于哪个设备，根据网关支持的设备自行定义
	union val_t old_val;  //变化上报后需要更新旧值
    union val_t new_val;  //从共享内存取出最新数据，放到new_val中
    int ret;  //默认为-1，采集成功后设置为0，采集失败再置-1
};


// struct student
// {
//     int num;
//     char name[64];
// };
struct msgbuf
{  // int flag;
    int mtype;
    char mdata[256];
};
static int *total;  //指向共享内存中数据节点总个数
static struct student *node_arr;  //指向共享内存中节点缓存数组头

#define MAX_NODE 50  //最大支持学生数目，实际可用是127
#define STD_NODE_LEN sizeof(struct student)
#define MAX_NODE_SIZE (MAX_NODE * STD_NODE_LEN)
//------

int flag1=2;
char buf[32];
char buf1[32];
char buf2[32];
 struct msgbuf send_buf;
 MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
void *handler_thread(void *arg)
 {
   
    
    sleep(4);
  
    while (1)
    {
      
    }
  
 }





volatile MQTTClient_deliveryToken deliveredtoken;

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char* payloadptr;

    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: ");

    payloadptr = message->payload;
cJSON *aaa=cJSON_Parse(payloadptr);

if(aaa == NULL)
    {
        printf("parse err\n");
        return -1;
    }
cJSON *eee=cJSON_GetObjectItem(aaa,"name");
 if (eee->valueint==1)
  {
   cJSON *ddd=cJSON_GetObjectItem(aaa,"key");
cJSON *bbb=cJSON_GetObjectItem(aaa,"bool");
send_buf.mtype = ddd->valueint;
strcpy(send_buf.mdata,bbb->valuestring);
for (int i = 0; i < 2; i++)
{
  /* code */


if(msg_queue_send("topic", &send_buf, sizeof(send_buf), 0) < 0)
        {
            printf("msg_queue_send error\n");
            return -1;
        }
printf("mtype= %d\n", send_buf.mtype);
printf("mdata= %s\n", send_buf.mdata);
  }
  }
if (eee->valueint==0)
{
  flag1=eee->valueint;
}
if (eee->valueint==2)
{
  flag1=eee->valueint;
}


    putchar('\n');
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

int main(int argc, char* argv[])
{
    
    
    int ret = -1;

    ret = shm_init(&para, "shm_test", MAX_NODE_SIZE);
    if(ret < 0)
    {
        return -1;
    }

    void *node_p = shm_getaddr(&para);
    if(node_p == NULL)
    {
        return -1;
    }

    memset(node_p, 0, MAX_NODE_SIZE);

    //前4个字节存储实际的学生数目
    total = (int *)node_p;
    //后面空间存储数据点
    node_arr = (struct student *)(node_p + sizeof(int));

    
    
    
     //创建线程
  pthread_t tid;
    if (pthread_create(&tid, NULL, handler_thread, NULL) != 0)
    {
        perror("pthread create err");
        return -1;
    }
    printf("main\n");

//往共享内存写点表
      FILE *file = fopen("node.json", "r");
    if(NULL == file)
    {
        perror("open err");
        return -1;
    }

    char buf[2024] = {0};
    //reads nmemb items of data, each size bytes long
    size_t len = fread(buf, 1, 2024, file);
    if(len < 0)
    {
        perror("read err");
        return -1;
    }

    // printf("%s\n", buf);
    //json的反序列化
    //把上面的原始字符串转换成一棵json树
    cJSON *root = cJSON_Parse(buf);
    if(NULL == root)
    {
        printf("parse err\n");
        return -1;
    }

    //获取root父节点上名字叫ver的子节点

    cJSON *item = cJSON_GetObjectItem(root, "version");
     // strcpy(node_arr->str,)
     strcpy(node_arr[0].str, item->valuestring);
   // printf("version = %s\n", item->valuestring);

    //获取root父节点上名字叫login的子节点
    cJSON *login = cJSON_GetObjectItem(root, "report");

    //获取login父节点上名字叫pwd的子节点
if (flag1==2)
{
   item = cJSON_GetObjectItem(login, "type");
 node_arr[1].type=item->valueint;
    item = cJSON_GetObjectItem(login, "period");
node_arr[1].dev_type=item->valueint;
}
if (flag1==0)
{
  
 node_arr[1].type=0;
   

}
   




cJSON *item1 = cJSON_GetObjectItem(root, "mqtt_server");
item = cJSON_GetObjectItem(item1, "addr");
strcpy(node_arr[2].str, item->valuestring);
item = cJSON_GetObjectItem(item1, "port");
node_arr[2].dev_type= item->valueint;


cJSON *item2 = cJSON_GetObjectItem(root, "mb_dev");
item = cJSON_GetObjectItem(item2, "addr");
strcpy(node_arr[3].str, item->valuestring);
item = cJSON_GetObjectItem(item2, "port");
node_arr[3].dev_type=item->valueint;


cJSON *item3 = cJSON_GetObjectItem(root, "mb_app");
item = cJSON_GetObjectItem(item3, "addr");
  strcpy(node_arr[4].str, item->valuestring);
item = cJSON_GetObjectItem(item3, "port");
node_arr[4].dev_type=item->valueint;


cJSON *item4 = cJSON_GetObjectItem(root, "stm32");
cJSON *item5 = cJSON_GetObjectItem(item4, "data");
 int	val= cJSON_GetArraySize(item5);
  int j=5;
for(int i=0;i<val;i++)
{  
   
  
cJSON * da=cJSON_GetArrayItem(item5, i);

item = cJSON_GetObjectItem(da, "key");
 node_arr[j].key=item->valueint;

item = cJSON_GetObjectItem(da, "name");
strcpy(node_arr[j].str, item->valuestring);

 item = cJSON_GetObjectItem(da, "type");
 node_arr[j].type=item->valueint;
   j++;
}

cJSON *item6 = cJSON_GetObjectItem(root, "m0");

cJSON *item7 = cJSON_GetObjectItem(item6, "name");
strcpy(node_arr[8].str, item7->valuestring);

cJSON *item8 = cJSON_GetObjectItem(item6, "data");
 int	vel= cJSON_GetArraySize(item8);
 int a=9;
for(int i=0;i<vel;i++)
{   
    
cJSON * da=cJSON_GetArrayItem(item8, i);

item = cJSON_GetObjectItem(da, "key");
 node_arr[a].key=item->valueint;

item = cJSON_GetObjectItem(da, "name");
 strcpy(node_arr[a].str, item->valuestring);


 item = cJSON_GetObjectItem(da, "type");
 node_arr[a].type=item->valueint;
 a++;
}


cJSON *item9 = cJSON_GetObjectItem(root, "modbus");

cJSON *item10 = cJSON_GetObjectItem(item9, "name");
strcpy(node_arr[15].str, item10->valuestring);

cJSON *item11= cJSON_GetObjectItem(item9, "data");
 int	vfl= cJSON_GetArraySize(item11);
 int b=16;
for(int i=0;i<vfl;i++)
{
cJSON * da=cJSON_GetArrayItem(item11, i);

item = cJSON_GetObjectItem(da, "key");
node_arr[b].key=item->valueint;

item = cJSON_GetObjectItem(da, "name");
 strcpy(node_arr[b].str, item->valuestring);

item = cJSON_GetObjectItem(da, "addr");
 node_arr[b].dev_type=item->valueint;

 item = cJSON_GetObjectItem(da, "type");
 node_arr[b].type=item->valueint;
 b++;
}

//21
cJSON *item12 = cJSON_GetObjectItem(root, "mbapp");

cJSON *item13 = cJSON_GetObjectItem(item12, "name");
 strcpy(node_arr[21].str, item13->valuestring);
cJSON *item14= cJSON_GetObjectItem(item12, "data");
 int	vgl= cJSON_GetArraySize(item14);
 int d=22;
for(int i=0;i<vgl;i++)
{
cJSON * da=cJSON_GetArrayItem(item14, i);

item = cJSON_GetObjectItem(da, "key");
 node_arr[d].key=item->valueint;

item = cJSON_GetObjectItem(da, "name");
 strcpy(node_arr[d].str, item->valuestring);

item = cJSON_GetObjectItem(da, "addr");
  node_arr[d].dev_type=item->valueint;

 item = cJSON_GetObjectItem(da, "type");
 node_arr[d].type=item->valueint;
 d++;
}


// -------
     

 
// 解析点表,并反序列化，放到共享内存里



// ------------

    MQTTClient client;
   
   
    int rc;
    int ch;

    MQTTClient_create(&client, ADDRESS, CLIENTID,
        MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
MQTTClient_message pubmsg = MQTTClient_message_initializer; // fa
    MQTTClient_deliveryToken token;  //fa
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
           "Press Q<Enter> to quit\n\n", TOPIC, CLIENTID, QOS);
    MQTTClient_subscribe(client, TOPIC, QOS);

    
    do 
    {
        ///ch = getchar();
   if (flag1==2)
   {
    sleep(8);
   }
   
      
       
        
cJSON *root=cJSON_CreateObject();

   


    
cJSON *cJSON_CreateString(const char *string);
cJSON_AddItemToObject(root, "version",  cJSON_CreateString(node_arr[0].str));


 cJSON * report=cJSON_CreateObject();
 cJSON_AddItemToObject(report, "type", cJSON_CreateNumber(node_arr[1].type));
  cJSON_AddItemToObject(report, "period", cJSON_CreateNumber(node_arr[1].dev_type));
cJSON_AddItemToObject(root, "report", report);

cJSON * repor=cJSON_CreateObject();
 cJSON_AddItemToObject(repor, "addr", cJSON_CreateString(node_arr[2].str));
  cJSON_AddItemToObject(repor, "port", cJSON_CreateNumber(node_arr[2].dev_type));
cJSON_AddItemToObject(root, "mqtt_server", repor);


cJSON * repo=cJSON_CreateObject();
 cJSON_AddItemToObject(repo, "addr", cJSON_CreateString(node_arr[3].str));
  cJSON_AddItemToObject(repo, "port", cJSON_CreateNumber(node_arr[3].dev_type));
cJSON_AddItemToObject(root, "mb_dev", repo);


cJSON * stm=cJSON_CreateObject();
cJSON *data=cJSON_CreateArray();
cJSON * stm1=cJSON_CreateObject();

 cJSON_AddItemToObject(stm1, "key", cJSON_CreateNumber(node_arr[9].key));
  cJSON_AddItemToObject(stm1, "name", cJSON_CreateString(node_arr[9].str));
cJSON_AddItemToObject(stm1, "type", cJSON_CreateNumber(node_arr[9].type));
cJSON_AddItemToObject(stm1, "old_data", cJSON_CreateNumber(node_arr[9].old_val.f_val));
cJSON_AddItemToObject(stm1, "new_data", cJSON_CreateNumber(node_arr[9].new_val.f_val));
cJSON_AddItemToArray(data,stm1);

cJSON * stm2=cJSON_CreateObject();
cJSON_AddItemToObject(stm2, "key", cJSON_CreateNumber(node_arr[10].key));
  cJSON_AddItemToObject(stm2, "name", cJSON_CreateString(node_arr[10].str));
cJSON_AddItemToObject(stm2, "type", cJSON_CreateNumber(node_arr[10].type));
cJSON_AddItemToObject(stm2, "old_data", cJSON_CreateNumber(node_arr[10].old_val.f_val));
cJSON_AddItemToObject(stm2, "new_data", cJSON_CreateNumber(node_arr[10].new_val.f_val));
cJSON_AddItemToArray(data,stm2);

cJSON * stm3=cJSON_CreateObject();
 cJSON_AddItemToObject(stm3, "key", cJSON_CreateNumber(node_arr[11].key));
  cJSON_AddItemToObject(stm3, "name", cJSON_CreateString(node_arr[11].str));
cJSON_AddItemToObject(stm3, "type", cJSON_CreateNumber(node_arr[11].type));
cJSON_AddItemToObject(stm3, "old_data", cJSON_CreateNumber(node_arr[11].old_val.f_val));
cJSON_AddItemToObject(stm3, "new_data", cJSON_CreateNumber(node_arr[11].new_val.f_val));
cJSON_AddItemToArray(data,stm3);

cJSON * stm4=cJSON_CreateObject();
 cJSON_AddItemToObject(stm4, "key", cJSON_CreateNumber(node_arr[12].key));
  cJSON_AddItemToObject(stm4, "name", cJSON_CreateString(node_arr[12].str));
cJSON_AddItemToObject(stm4, "type", cJSON_CreateNumber(node_arr[12].type));
cJSON_AddItemToObject(stm4, "old_data", cJSON_CreateNumber(node_arr[12].old_val.f_val));
cJSON_AddItemToObject(stm4, "new_data", cJSON_CreateNumber(node_arr[12].new_val.f_val));
cJSON_AddItemToArray(data,stm4);

cJSON * stm5=cJSON_CreateObject();
 cJSON_AddItemToObject(stm5, "key", cJSON_CreateNumber(node_arr[13].key));
  cJSON_AddItemToObject(stm5, "name", cJSON_CreateString(node_arr[13].str));
cJSON_AddItemToObject(stm5, "type", cJSON_CreateNumber(node_arr[13].type));
cJSON_AddItemToObject(stm5, "old_data", cJSON_CreateNumber(node_arr[13].old_val.f_val));
cJSON_AddItemToObject(stm5, "new_data", cJSON_CreateNumber(node_arr[13].new_val.f_val));
cJSON_AddItemToArray(data,stm5);

cJSON * stm6=cJSON_CreateObject();
 cJSON_AddItemToObject(stm6, "key", cJSON_CreateNumber(node_arr[14].key));
  cJSON_AddItemToObject(stm6, "name", cJSON_CreateString(node_arr[14].str));
cJSON_AddItemToObject(stm6, "type", cJSON_CreateNumber(node_arr[14].type));
cJSON_AddItemToObject(stm6, "old_data", cJSON_CreateNumber(node_arr[14].old_val.f_val));
cJSON_AddItemToObject(stm6, "new_data", cJSON_CreateNumber(node_arr[14].new_val.f_val));
cJSON_AddItemToArray(data,stm6);

cJSON_AddItemToObject(stm,"data",data);
cJSON_AddItemToObject(root,"stm",stm);

cJSON * modbus=cJSON_CreateObject();
cJSON *data2=cJSON_CreateArray();
cJSON * mo1=cJSON_CreateObject();

 cJSON_AddItemToObject(mo1, "key", cJSON_CreateNumber(node_arr[16].key));
  cJSON_AddItemToObject(mo1, "name", cJSON_CreateString(node_arr[16].str));
cJSON_AddItemToObject(mo1, "type", cJSON_CreateNumber(node_arr[16].type));
cJSON_AddItemToObject(mo1, "old_data", cJSON_CreateNumber(node_arr[16].old_val.f_val));
cJSON_AddItemToObject(mo1, "new_data", cJSON_CreateNumber(node_arr[16].new_val.f_val));
cJSON_AddItemToArray(data2,mo1);


cJSON * mo2=cJSON_CreateObject();
 cJSON_AddItemToObject(mo2, "key", cJSON_CreateNumber(node_arr[17].key));
  cJSON_AddItemToObject(mo2, "name", cJSON_CreateString(node_arr[17].str));
cJSON_AddItemToObject(mo2, "type", cJSON_CreateNumber(node_arr[17].type));
cJSON_AddItemToObject(mo2, "old_data", cJSON_CreateNumber(node_arr[17].old_val.f_val));
cJSON_AddItemToObject(mo2, "new_data", cJSON_CreateNumber(node_arr[17].new_val.f_val));
cJSON_AddItemToArray(data2,mo2);

cJSON * mo3=cJSON_CreateObject();
 cJSON_AddItemToObject(mo3, "key", cJSON_CreateNumber(node_arr[18].key));
  cJSON_AddItemToObject(mo3, "name", cJSON_CreateString(node_arr[18].str));
cJSON_AddItemToObject(mo3, "type", cJSON_CreateNumber(node_arr[18].type));
cJSON_AddItemToObject(mo3, "old_data", cJSON_CreateNumber(node_arr[18].old_val.f_val));
cJSON_AddItemToObject(mo3, "new_data", cJSON_CreateNumber(node_arr[18].new_val.f_val));
cJSON_AddItemToArray(data2,mo3);

cJSON * mo4=cJSON_CreateObject();
 cJSON_AddItemToObject(mo4, "key", cJSON_CreateNumber(node_arr[19].key));
  cJSON_AddItemToObject(mo4, "name", cJSON_CreateString(node_arr[19].str));
cJSON_AddItemToObject(mo4, "type", cJSON_CreateNumber(node_arr[19].type));
cJSON_AddItemToObject(mo4, "old_data", cJSON_CreateNumber(node_arr[19].old_val.f_val));
cJSON_AddItemToObject(mo4, "new_data", cJSON_CreateNumber(node_arr[19].new_val.f_val));
cJSON_AddItemToArray(data2,mo4);

cJSON * mo5=cJSON_CreateObject();
 cJSON_AddItemToObject(mo5, "key", cJSON_CreateNumber(node_arr[20].key));
  cJSON_AddItemToObject(mo5, "name", cJSON_CreateString(node_arr[20].str));
cJSON_AddItemToObject(mo5, "type", cJSON_CreateNumber(node_arr[20].type));
cJSON_AddItemToObject(mo5, "old_data", cJSON_CreateNumber(node_arr[20].old_val.f_val));
cJSON_AddItemToObject(mo5, "new_data", cJSON_CreateNumber(node_arr[20].new_val.f_val));
cJSON_AddItemToArray(data2,mo5);
cJSON_AddItemToObject(modbus,"data2",data2);
cJSON_AddItemToObject(root,"modbus",modbus);

printf("%f\n",node_arr[17].new_val.f_val);


char  *ccc=cJSON_Print(root);
   pubmsg.payload = ccc;
    pubmsg.payloadlen = (int)strlen(ccc);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
if(flag1==2)
{

 MQTTClient_publishMessage(client, TOPIC1, &pubmsg, &token);
    printf("Waiting for up to %d seconds for publication of %s\n"
            "on topic %s for client with ClientID: %s\n",
            (int)(TIMEOUT/1000), PAYLOAD, TOPIC1, CLIENTID);
    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);

}
   if (flag1==0)
   {
    sleep(1);
 for (int i = 9; i <13; i++)
 {
   if(node_arr[i].old_val.f_val!=node_arr[i].new_val.f_val)
{
MQTTClient_publishMessage(client, TOPIC1, &pubmsg, &token);
    printf("Waiting for up to %d seconds for publication of %s\n"
            "on topic %s for client with ClientID: %s\n",
            (int)(TIMEOUT/1000), PAYLOAD, TOPIC1, CLIENTID);
    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
}
 }

 
for (int i = 16; i <=19; i++)
 {
   if(node_arr[i].old_val.f_val!=node_arr[i].new_val.f_val)
{
MQTTClient_publishMessage(client, TOPIC1, &pubmsg, &token);
    printf("Waiting for up to %d seconds for publication of %s\n"
            "on topic %s for client with ClientID: %s\n",
            (int)(TIMEOUT/1000), PAYLOAD, TOPIC1, CLIENTID);
    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
}
 }


   
for (int i = 9; i <13; i++)
 {
   node_arr[i].old_val.f_val=node_arr[i].new_val.f_val;

   }
   
for (int i = 16; i <=19; i++)
 {
   node_arr[i].old_val.f_val=node_arr[i].new_val.f_val;

   }


   }

    } while(ch!='Q' && ch != 'q');

    MQTTClient_unsubscribe(client, TOPIC);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    return rc;
}
